package com.lagou.rpc.consumer;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@SpringBootApplication
public class ClientBootStrapApplication {
    public static String PATH = "/provider";

    private static ConcurrentHashMap<String, ZkThread> servers = new ConcurrentHashMap<>();
    private static CuratorFramework client;

    public static void main(String[] args) throws Exception {
        SpringApplication.run(ClientBootStrapApplication.class, args);
        //获取连接
        client = getCuratorClient();
        //获取节点
        List<String> serverList = client.getChildren().forPath(PATH);
        for (int i = 0; i < serverList.size(); i++) {
            addServer(serverList.get(i));
        }
        //监听逻辑
        zkWatch(client, PATH);
    }

    /**
     * 注册监听
     * TreeCache: 可以将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，
     * 呈现树形目录的监听，可以设置监听深度，最大监听深度为 int 类型的最大值。
     */
    private static void zkWatch(final CuratorFramework client, final String path) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);
        treeCache.getListenable().addListener(new TreeCacheListener() {
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData eventData = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED: {
                        System.out.println(path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
                        String path = eventData.getPath();
                        final String host = path.substring(path.lastIndexOf("/") + 1);
                        if (!host.equals(PATH.substring(PATH.lastIndexOf("/") + 1))) {
                            addServer(host);
                        }
                    }
                    break;
                    case NODE_UPDATED: {
                        System.out.println(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
                    }
                    break;
                    case NODE_REMOVED: {
                        System.out.println(eventData.getPath() + "节点被删除");
                        String path = eventData.getPath();
                        String host = path.substring(path.lastIndexOf("/") + 1);
                        removeServer(host);
                    }
                    break;
                    default:
                        break;
                }
            }
        });
        treeCache.start();
        while (true) {
            Thread.sleep(3000);
        }
    }

    public static CuratorFramework getCuratorClient() {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curatorClient = CuratorFrameworkFactory.builder()
                .connectString("192.168.204.128:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("base")
                .build();
        curatorClient.start();
        return curatorClient;
    }

    private static void addServer(String host) throws Exception {
        if (!servers.containsKey(host)) {
            ZkThread zkThread = new ZkThread(host);
            zkThread.start();
            servers.put(host, zkThread);
        }
    }

    private static void removeServer(String host) {
        if (servers.containsKey(host)) {
            ZkThread zkThread = servers.get(host);
            zkThread.exit();
            servers.remove(host);
        }
    }

    public static CuratorFramework getClient() {
        if (client == null) {
            client = getCuratorClient();
        }
        return client;
    }

    public static void updateNodeData(String nodeName, String data) throws Exception {
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(nodeName);
        client.setData().withVersion(stat.getVersion()).forPath(nodeName, String.valueOf(data).getBytes());
    }

    private static class ZkThread extends Thread {
        public volatile boolean exit = true;
        private String host;

        public ZkThread(String host) {
            this.host = host;
        }

        public void exit() {
            this.exit = false;
        }

        @Override
        public void run() {
            super.run();
            while (exit) {
                super.run();
                if (Thread.currentThread().isInterrupted()) {
                    System.out.println(host + " - 线程终止");
                    break;
                }
                String nodeName = PATH + "/" + host;
                try {
                    byte[] bytes = client.getData().forPath(nodeName);
                    String data = new String(bytes);
                    System.out.println(host + " - 获取到的节点数据内容： " + data);
                    long nodeTime = Long.parseLong(data);
                    long currentTimeMillis = System.currentTimeMillis();
                    if (nodeTime == 0) {
                        System.out.println(host + " - 节点内容为空");
                    } else if (currentTimeMillis - nodeTime > 5000) {
                        System.out.println(host + " - 超时，清除节点内容");
                        updateNodeData(nodeName, String.valueOf(0));
                    } else {
                        System.out.println(host + " - 未超时");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
